package com.ruoyi.socket.socketserver;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.socket.dto.SocketDto;
import com.ruoyi.socket.dto.WsSubBO;
import com.ruoyi.socket.manager.WebSocketUserManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
@ServerEndpoint("/ws/{userId}")
public class WebSocketServers {
    // 原子化在线计数器
    private static final AtomicInteger onlineCount = new AtomicInteger(0);

    // 线程安全集合初始化
    public static final Map<String, List<String>> detailMap = new ConcurrentHashMap<>();
    public static final Map<String, List<String>> tradeMap = new ConcurrentHashMap<>();
    public static final Map<String, List<String>> klineMap = new ConcurrentHashMap<>();

    static {
        detailMap.put("DETAIL", new CopyOnWriteArrayList<>());
        tradeMap.put("TRADE", new CopyOnWriteArrayList<>());
        klineMap.put("KLINE", new CopyOnWriteArrayList<>());
    }

    // 连接管理
    private static final Map<String, WebSocketServers> webSocketMap = new ConcurrentHashMap<>();

    // Spring依赖静态注入
    private static RedisCache redisCache;
    private static WebSocketUserManager webSocketUserManager;

    @Autowired
    public void setRedisCache(RedisCache redisCache) {
        WebSocketServers.redisCache = redisCache;
    }

    @Autowired
    public void setWebSocketUserManager(WebSocketUserManager webSocketUserManager) {
        WebSocketServers.webSocketUserManager = webSocketUserManager;
    }

    private Session session;
    private String userId;

    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        handleNewConnection(session, userId);
        sendConnectionAck();
    }

    private void handleNewConnection(Session session, String userId) {
        this.session = session;
        this.userId = userId;

        // 注册到 UserManager，替代原本 webSocketMap 里的 session 管理
        if (webSocketUserManager != null) {
//            webSocketUserManager.(userId, session);
        }

        webSocketMap.compute(userId, (k, existing) -> {
            if (existing != null) {
                closeExistingConnection(existing);
            }
            onlineCount.incrementAndGet();
            return this;
        });

        log.info("Connection established: {}, Online: {}", userId, onlineCount.get());
    }

    private void closeExistingConnection(WebSocketServers existing) {
        try {
            existing.session.close();
            log.debug("Closed duplicate connection: {}", existing.userId);
        } catch (IOException e) {
            log.debug("Error closing duplicate connection: {}", e.getMessage());
        }
    }

    private void sendConnectionAck() {
        if (isConnectionActive()) {
            String ackMessage = JSONObject.toJSONString(new SocketDto());
            log.info("Sending ACK to {}: {}", userId, ackMessage);
            sendMessage(ackMessage);
        } else {
            log.error("Session not active for ACK: {}", userId);
        }
    }

    @OnClose
    public void onClose() {
        webSocketMap.computeIfPresent(userId, (k, v) -> {
            cleanupResources();
            return null;
        });
        log.info("Connection closed: {}, Online: {}", userId, onlineCount.get());
    }

    private void cleanupResources() {
        onlineCount.decrementAndGet();

        if (redisCache != null) {
            redisCache.deleteObject(userId);
        }

        removeFromAllSubscriptions();
        closeSessionQuietly();
    }

    private void removeFromAllSubscriptions() {
        safeRemove(detailMap.get("DETAIL"), "DETAIL");
        klineMap.values().forEach(list -> safeRemove(list, "KLINE"));
        tradeMap.values().forEach(list -> safeRemove(list, "TRADE"));
    }

    private void safeRemove(List<String> list, String type) {
        if (list != null) {
            synchronized (list) {
                if (list.remove(userId)) {
                    log.debug("Removed {} subscription for: {}", type, userId);
                }
            }
        }
    }

    private void closeSessionQuietly() {
        try {
            if (session != null && session.isOpen()) {
                session.close();
            }
        } catch (IOException e) {
            log.debug("Session closure error: {}", e.getMessage());
        }
    }

    @OnMessage
    public void onMessage(String message, Session session) {
        if (isValidMessage(message)) {
            processSubscription(message);
        }
    }

    private boolean isValidMessage(String message) {
        return StringUtils.isNotBlank(message) && !"heartbeat".equals(message);
    }

    private void processSubscription(String message) {
        try {
            WsSubBO subscription = parseSubscription(message);
            if (webSocketUserManager != null) {
                webSocketUserManager.subscribeMsg(subscription);
            }
        } catch (Exception e) {
            log.error("Subscription processing failed: {}", e.getMessage());
        }
    }

    private WsSubBO parseSubscription(String message) {
        WsSubBO subscription = JSON.parseObject(message).toJavaObject(WsSubBO.class);
        subscription.setUserId(userId);
        return subscription;
    }

    @OnError
    public void onError(Session session, Throwable error) {
        log.error("Connection error [{}]: {}", userId, error.getMessage());
        cleanupResources();
    }

    public  void sendMessage(String message) {

//        if (session != null && session.isOpen()) {
//            this.session.getAsyncRemote().sendText(message, result -> {
//                if (!result.isOK()) {
//                    log.error("异步发送失败11: {}", result.getException().getMessage());
//                } else {
//                    log.info("异步处理111");
//                }
//            });
//        }

        try {
            synchronized (session) { // 避免多线程冲突
                session.getBasicRemote().sendText(message);
            }
        } catch (Exception e) {
            log.error("发送消息给 {} 失败: {} 消息：{}", userId, e.getMessage(), message);
        }

    }

    private boolean isConnectionActive() {
        return session != null && session.isOpen();
    }

    private void handleSendError(Exception e) {
        if (e instanceof IOException) {
            String errorMsg = e.getMessage();
            if (errorMsg.contains("Broken pipe") || errorMsg.contains("Connection reset")) {
                log.debug("Client {} disconnected abruptly: {}", userId, errorMsg);
            } else {
                log.error("Failed to send message to {}: {}", userId, errorMsg);
            }
        } else {
            log.error("Unexpected error sending message to {}: {}", userId, e.getMessage());
        }
        // 安全移除并清理
        webSocketMap.remove(userId);
        cleanupResources();
    }

    public static void sendInfo(String message, String userId) {
        WebSocketServers client = webSocketMap.get(userId);
        if (client != null) {
            client.sendMessage(message);
        }
    }

    public static int getOnlineCount() {
        return onlineCount.get();
    }
}